use core::future::Future;
use core::pin::Pin;
use futures_util::stream::{unfold, Unfold};
use soketto::connection::{Error as WsError, Receiver};
use soketto::Data;
use tokio::net::TcpStream;
use tokio_util::compat::Compat;

pub const PROTOCOL: &'static str = "unmp";
pub const PATH: &'static str = "/unmp";
pub const DATA_LEN: usize = 4096;

type WsRx = Receiver<Compat<TcpStream>>;
type WsFut = Pin<
    Box<dyn Future<Output = Option<(Result<(Data, Vec<u8>), WsError>, WsRx)>> + Send + 'static>,
>;
pub type Rx = Unfold<WsRx, fn(WsRx) -> WsFut, WsFut>;

fn ws_recv(mut rx: WsRx) -> WsFut {
    Box::pin(async move {
        let mut data = Vec::<u8>::with_capacity(DATA_LEN);
        let result = rx.receive_data(&mut data).await;
        let ret = match result {
            Ok(ty) => Ok((ty, data)),
            Err(err) => Err(err),
        };
        Some((ret, rx))
    })
}
pub fn convert_rx(rx: WsRx) -> Rx {
    unfold::<_, fn(WsRx) -> WsFut, _, _>(rx, ws_recv)
}
